package com.dream.cms.controller;

import com.dream.cms.config.KafkaTopicProperties;
import com.dream.cms.message.PracticalAdvice;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
//import org.springframework.web.bind.annotation.GetMapping;
//import org.springframework.web.bind.annotation.ResponseBody;
//import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

@Slf4j
@RestController
public class HelloKafkaController {

//    @Autowired
//    private KafkaTemplate kafkaTemplate;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
//    private final String topicName;
//    private final int messagesPerRequest;
    private CountDownLatch latch;


    @Autowired
    private KafkaTopicProperties kafkaTopicProperties;

//    public HelloKafkaController(
//            final KafkaTemplate<String, Object> template) {
//        this.template = template;
//        this.topicName = kafkaTopicProperties.getTopicName();
//        this.messagesPerRequest = kafkaTopicProperties.getMessagesPerRequest();
//    }

    @GetMapping("/hello")
    public String hello() throws Exception {
//        latch = new CountDownLatch(kafkaTopicProperties.getMessagesPerRequest());
//        IntStream.range(0, kafkaTopicProperties.getMessagesPerRequest())
//                .forEach(i -> this.template.send(kafkaTopicProperties.getTopicName(),"kafka"+i)
//                );
//        latch.await(60, TimeUnit.SECONDS);
        ListenableFuture<SendResult<String,Object>> fulter = this.kafkaTemplate.send(kafkaTopicProperties.getTopicName(),"1234");
        fulter.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("发送失败");
            }

            @Override
            public void onSuccess(SendResult<String, Object> integerStringSendResult) {
                System.out.println("发送成功");
            }
        });
        log.info("All messages received");
        return "Hello Kafka!";
    }
}
